package com.xiongmm.stream.sink;

import com.xiongmm.stream.database.DataManipulation;
import com.xiongmm.stream.domain.InstanceInfoAllVo;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

/**
 * @author: XiongMM
 * @date: 2019/5/30 16:49
 * @description:批量插入数据
 */
public class SinkInfoAllToMySQL extends RichSinkFunction<List<InstanceInfoAllVo>> {

    private static final Logger LOGGER = LoggerFactory.getLogger(SinkInfoCmdToMySQL.class);

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
    }

    @Override
    public void invoke(List<InstanceInfoAllVo> records, Context context) throws Exception {

        try {
            for (InstanceInfoAllVo infoAllVo : records) {
                //插入cache_app_statistic_log_new
                DataManipulation.insertInstanceInfoAll(infoAllVo);
                //如果不存在就插入cache_instance_stat_new,存在就更新
                DataManipulation.insertInstanceInfoStat(infoAllVo);
            }
        } catch (Exception ex) {
            LOGGER.error("SinkInfoAllToMySQL.execute ------> 插入info all失败～～～", ex);
        }
    }

    @Override
    public void close() throws Exception {
        super.close();
    }
}
